package com.example.rocket.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 监听事务类型的消息
 * @Author ifredomvip@gmail.com
 * @Date 2022/8/16 16:45
 * @Version 1.0.0
 * @Description
 **/
@Slf4j
@RocketMQTransactionListener
public class TransactionListener implements RocketMQLocalTransactionListener {

    private ConcurrentHashMap<Object, String> localTrans = new ConcurrentHashMap<>();

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        // message: org.springframework.messaging.Message, 是Spring自己封装的类, 和RocketMQ的Message不是一个类, 里面没有tags/keys等内容

        Object id = message.getHeaders().get("id");
        String destination = o.toString();
        assert id != null;
        localTrans.put(id, destination);


        log.info("executeLocalTransaction message:{}, object:{}", message, o);
        log.info("payload: {}", new String((byte[]) message.getPayload()));
        MessageHeaders headers = message.getHeaders();
        log.info("tags: {}", headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TAGS));
        log.info("rocketmq_TOPIC: {}", headers.get("rocketmq_TOPIC"));
        log.info("rocketmq_QUEUE_ID: {}", headers.get("rocketmq_QUEUE_ID"));
        log.info("rocketmq_MESSAGE_ID: {}", headers.get("rocketmq_MESSAGE_ID"));
        log.info("rocketmq_TRANSACTION_ID: {}", headers.get("rocketmq_TRANSACTION_ID"));
        log.info("TRANSACTION_CHECK_TIMES: {}", headers.get("TRANSACTION_CHECK_TIMES"));
        log.info("id: {}", headers.get("id"));

        return RocketMQLocalTransactionState.UNKNOWN;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {

        // SpringBoot的消息对象中，并没有transactionId这个属性。跟原生API不一样。

        // String destination = localTrans.get(msg.getTransactionId());

        // 在调用了checkLocalTransaction后, 另一个常规消息监听器才能收到消息
        return RocketMQLocalTransactionState.COMMIT;

    }
}
